package com.example.springboot.controller.bigfileinput;

import com.google.common.collect.Lists;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.LineIterator;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 10 个线程，每个线程读取1000行数据
 *
 * @Author:ChenZhangKun
 * @Date: 2021/12/19 14:37
 */
@RestController
@RequestMapping("/czk/file")
@Api(tags = "多线程批量导入大文件")
public class BigFileInputController {
    private ExecutorService threadPool = Executors.newFixedThreadPool(10);

    @GetMapping("/input")
    @ApiOperation(value = "导入文件")
    public String inputFile() {
        // 按行读取
        try (LineIterator iterator = IOUtils.lineIterator(new FileInputStream("test.txt"), "utf-8")) {
            // 存储执行任务的行数
            List<String> line = new ArrayList<>();
            // 存储异步任务
            List<ConvertTask> task = new ArrayList<>();
            while (iterator.hasNext()) {
                // 拿到下一行
                String next = iterator.next();
                // 加入
                line.add(next);
                // 设置每个线程执行的行数
                if (line.size() == 1000) {
                    // 新建任务
                    task.add(new ConvertTask(line));
                }
                if (task.size() == 10) {
                    // 执行任务
                    asyncBatchExecuteTask(task);
                    // 清空
                    line.clear();
                }
            }
            // 文件读取结束，但是可能还存在未被内容
            task.add(new ConvertTask(Lists.newArrayList(line)));
            // 最后再执行一次
            asyncBatchExecuteTask(task);
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        } finally {

        }
        return "保存成功";
    }

    private void asyncBatchExecuteTask(List<ConvertTask> tasks) throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(tasks.size());
        for (ConvertTask task : tasks) {
            task.setCountDownLatch(countDownLatch);
            threadPool.submit(task);
        }
        // 主线程等待异步线程 countDownLatch 执行结束
        countDownLatch.await();
        // 清空，重新添加任务
        tasks.clear();
    }

    private static class ConvertTask implements Runnable {
        private CountDownLatch countDownLatch;
        private List<String> line;

        public ConvertTask(List<String> line) {
            this.line = line;
        }

        public void setCountDownLatch(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                for (String s : line) {
                    // 导入数据库
                    System.out.println("导入数据库" + s);
                }
            } finally {
                countDownLatch.countDown();
            }

        }
    }
}
